package com.atguigu.chapter05.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.ArrayList;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/17 10:44
 */
public class Flink01_Sink_Redis_Hash {
    public static void main(String[] args) throws Exception {
        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
        waterSensors.add(new WaterSensor("传感器", 1607527995000L, 30));
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        // redis: string list set hash zset
        
        // 存储每个传感器水位的最大值
        /*
        key             value
        sensor_1        {"id": "sensor_1", "ts": ..., "vc": ....}
        
        
        
        hash
        
        key             field           value
        传感器           sensor_1       字符串格式的json
        
         */
        
        FlinkJedisConfigBase conf = new FlinkJedisPoolConfig.Builder()
            .setHost("hadoop162")
            .setPort(6379)
            .setDatabase(0)
            .setMaxTotal(100)
            .setMinIdle(10)
            .setTimeout(10 * 1000)
            .setMinIdle(1)
            .build();
        env
            .fromCollection(waterSensors)
            .keyBy(WaterSensor::getId)
            .max("vc")
            .addSink(new RedisSink<>(conf, new RedisMapper<WaterSensor>() {
                
                @Override
                public RedisCommandDescription getCommandDescription() {
                    // 参数1: 根据要写的数据类型, 选择对应的命令
                    // 参数2: 只针对hash和zset有效, 其他的数据结构忽略
                    // hash:   key    value(map)
                    //          h1     field            value
                    //                   a                  97
                    return new RedisCommandDescription(RedisCommand.HSET, "sensor");
                }
                
                // 返回要写的这条数据的key
                @Override
                public String getKeyFromData(WaterSensor data) {
                    return data.getId();
                }
                
                @Override
                public String getValueFromData(WaterSensor data) {
                    return JSON.toJSONString(data);
                }
            }));
        
        env.execute();
        
    }
}
